<?php namespace xlx;

use \Workerman\Events\EventInterface;
use \Workerman\Connection\TcpConnection;
use \Workerman\Worker;

class TcpPool extends \Workerman\Worker
{

 

    //全局计算机唯一
    public static $addr = 'tcp://127.0.0.1:33336';
    //全局计算机空闲进程 
    public static $conns=[];
    //工作进程链接pid
    public static $links=[];
    public static function worker()
    {
        $CLASS = \Workerman\Worker::getEventLoopName();
        $event = \Workerman\Worker::$globalEvent = new $CLASS;
        $client = \stream_socket_client(self::$addr, $errno, $errstr) or exit($errstr);
        $client_addr = \stream_socket_get_name($client, false);
        $server = \stream_socket_server($client_addr, $errno, $errstr) or exit($errstr);
        \stream_set_blocking($server, false);
        $event->add($server, EventInterface::EV_READ, function ($socket) {
            $client = \stream_socket_accept($socket, 0, $addr);
            $conn = new TcpConnection($client);
            $conn->protocol='\xlx\Protocols\Frame';
            $conn->onMessage=function ($conn, $data) {
                switch ($data->act) {
                    case 'send':
                        $mailer = Actor::getMailer($data->pid);
                        $mailer->msgbox[] = $data->msg;
                        $mailer->resolve();
                        break;
                    case 'link':
                        self::$links[$data->pid][] = $data->link;
                        break;
                    case 'throw':   
                        $mailer = Actor::getMailer($data->pid); 
                        $mailer->reject(new \Exception($data->message));
                        // $mailer->msgbox[] = $data->msg;
                        // $mailer->resolve();
                        break;
                    default:
                        echo \json_encode($data);
                        echo "!\n";
                        break;
                }
            };
        });
        $conn = new TcpConnection($client);
        $conn->protocol='\xlx\Protocols\Frame';
        $conn->onMessage=function ($conn, $data) {
            try {
                switch ($data->act) {
                    case 'start':
                        break;
                    case 'eval':
                        eval($data->code);
                        break;
                    case 'spawn':
                        if (isset($data->link)) {
                            self::$links[$data->pid][] = $data->link;
                        }
                        $opisclosure = \unserialize($data->code);
                        $opisclosure->__PID__ = $data->pid;
                        \xlx\co($opisclosure, $data->pid);
                        break;
                    case 'die':
                        echo 'die die die';
                        break;
                    default:
                        echo "\n".\json_encode($data);
                        echo "!\n";
                        break;
                }
                $conn->send(['act'=>'idle']);
            } catch (\Throwable $e) {
                if (empty(self::$links[$data->pid])) {
                    return;
                }
                foreach (self::$links[$data->pid] as $pid) {
                    if (\preg_match('/([\d\.]+:\d+):(\d+)/', $pid, $arr)) { 
                        self::postData($arr[1], [
                            'act'=>'throw',
                            'pid'=>$pid,//$data->pid, 
                            'message'=>$e->getMessage(),
                            // 'file'=>$e->getFile(),
                            // 'line'=>$e->getLine()
                        ]);
                    }
                }
            }
        };
        $conn->onClose=function () {
            throw new \Exception("Error Processing Request", 1);
        };
        $conn->send(['act'=>'idle']);
        $event->loop();
    }
    public static function server($socket)
    {
        $client = \stream_socket_accept($socket, 0, $addr);
        $conn = new TcpConnection($client, $addr);
        $conn->addr = $addr;
        $conn->protocol='\xlx\Protocols\Frame';
        $conn->onMessage=function (TcpConnection $conn, $data) {
            switch ($data->act) {
                case 'eval':
                    //same
                case 'start':
                    //same
                case 'spawn': 
                    $addr = array_rand(self::$conns);
                    $data->pid = "$addr:".(int)(microtime(true)*1000);
                    usleep(7);//唯一
                    self::$conns[$addr]->send($data);
                    $conn->send($data->pid); 
                    break;
                case 'register':
                    break;

                case 'unregister':
                    break;
                    
                case 'whereis':
                    break;

                case 'idle': 
                    self::$conns[$conn->addr]=$conn;
                    break;
                case 'throw':
                    echo "\nSever PID:{$data->pid} throw";
                    break;
                case 'exit':
                    exit("server was dead \n");
                    break;
                    
                default:
                    echo \json_encode($data);
                    echo "!\n";
                    break;
            }
        };
        $conn->onClose=function () use ($conn) {
            unset(self::$conns[$conn->addr]); 
        };
    }

    public static function startServer()
    {
        $CLASS = \Workerman\Worker::getEventLoopName();
        $event = \Workerman\Worker::$globalEvent = new $CLASS;
        if ($socket = @\stream_socket_client(self::$addr)) {
            \fwrite($socket, \xlx\Protocols\Frame::encode(['act'=>'exit']));
            //throw new \Exception("Server Has Running", 1);
        }
        $socket = \stream_socket_server(self::$addr, $errno, $errstr) or exit($errstr);
        \stream_set_blocking($socket, false);
        $event->add($socket, EventInterface::EV_READ, function ($socket) {
            self::server($socket);
        });
        self::startWorker();
        if (extension_loaded('xdebug')) {
            echo "xdebug is running\n";
        } else {
            $event->add(1, EventInterface::EV_TIMER, function () { 
                $conntotal = count(self::$conns);
                if ( $conntotal<4 ) { 
                    self::startWorker();
                }
                echo "PROC:  $conntotal \n"; 
            });
        }
        $event->loop();
    }
    public static function startWorker()
    {
        if (strpos(strtolower(PHP_OS), 'win') === 0) {
            static $include;
            if (empty($include)) {
                foreach (get_declared_classes() as $composer) {
                    if (substr($composer, 0, 22)=='ComposerAutoloaderInit') {
                        $ref = new \ReflectionClass($composer);
                        $file = $ref->getFileName();
                        $class = xlx\OTPWorker::class;
                        $include = "require_once '$file';$composer::getLoader();";
                    }
                }
            }
            if (empty($include)) {
                throw new Exception("Error Processing Request", 1);
            }
            popen('start /B '.\PHP_BINARY.' -r "'.$include
                        .'class_alias(\xlx\Windows\Events\Select::class, \Workerman\Events\Select::class);'
                        .__CLASS__.'::worker();"', 'r' );
        } else {
            if (!\pcntl_fork()) {
                static::worker();
            }
        }
    }
    //工作进程链接
    static $conn;//pool
    public static function postData($ip, $data, $return = false, $timeout = 5*1000)
    {
        if (empty(self::$conn[$ip])) {
            self::$conn[$ip] = \stream_socket_client($ip, $errno, $errstr) or exit($errstr);
            \stream_set_blocking(self::$conn[$ip], false);
        }
        \fwrite(self::$conn[$ip], \xlx\Protocols\Frame::encode($data));
        if ($return) {
            do {
                $rs =  [self::$conn[$ip]];
                $ws = $es = [];
            } while (!\stream_select($rs, $ws, $es, $timeout)); //block
            $result = \xlx\Protocols\Frame::decode(stream_get_contents($rs[0]));
            return $result;
        }
    }
}
